package com.lcy.struct.bTreePlus.disk.recovery.redo;

import com.alibaba.fastjson.JSONObject;
import com.lcy.struct.bTreePlus.disk.constants.BPlusPublicConstants;
import com.lcy.struct.bTreePlus.disk.core.BPlusTree;
import com.lcy.struct.bTreePlus.disk.core.BPlusTreeFactory;
import com.lcy.struct.bTreePlus.disk.virtual.VirtualBPlus;
import com.lcy.struct.bTreePlus.entity.RedoInfo;
import com.lcy.struct.bTreePlus.property.BPlusFileInfo;
import lombok.SneakyThrows;

import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author lucongyang
 * @date Created in  10:26
 * @description
 */
class DoRedo {
    private static ReentrantReadWriteLock LOCK = new ReentrantReadWriteLock();
    private static final Lock WRITE_LOCK = LOCK.writeLock();
    private static final Lock READ_LOCK = LOCK.readLock();

    /**
     * 执行redo日志里未持久化的事务
     */
    @SneakyThrows
    List<RedoInfo> getTransactionRedoList() {
        FileChannel channel = RedoCenter.getInstance().getRedoChannel();
        long position = RedoCenter.getInstance().getRedoEndPosition();
        boolean straightForward = false;
        boolean needRedo = false;
        List<List<byte[]>> toDispose = new ArrayList<>();
        List<byte[]> tempList = new ArrayList<>();
        int n = Math.max(BPlusPublicConstants.HEAD_SIZE, BPlusPublicConstants.REDO_PAGE_SIZE);
        while (position > n) {
            long commitEndStart = position - BPlusPublicConstants.REDO_PAGE_SIZE;
            int redoContentStrLength = BPlusPublicConstants.REDO_CONTENT.getBytes().length;
            int commitEndStrSize = Math.max(redoContentStrLength,
                    BPlusPublicConstants.REDO_END_STR.getBytes().length + BPlusPublicConstants.REDO_END_COMMIT_STR.getBytes().length);
            MappedByteBuffer mappedByteBuffer = channel.map(FileChannel.MapMode.READ_ONLY, commitEndStart, commitEndStrSize);
            byte[] redoCommitEndBytes = new byte[commitEndStrSize];
            mappedByteBuffer.get(redoCommitEndBytes);
            String redoCommitEndLengthStr = new String(redoCommitEndBytes).trim();

            if (redoCommitEndLengthStr.startsWith(BPlusPublicConstants.REDO_CONTENT)) {
                if (straightForward) {
                    position -= BPlusPublicConstants.REDO_PAGE_SIZE;
                    continue;
                }
                if (needRedo) {
                    long start = position - BPlusPublicConstants.REDO_PAGE_SIZE + redoContentStrLength;
                    MappedByteBuffer buffer = channel.map(FileChannel.MapMode.READ_ONLY, start, BPlusPublicConstants.REDO_PAGE_SIZE - redoContentStrLength);
                    byte[] bytes = new byte[BPlusPublicConstants.REDO_PAGE_SIZE - redoContentStrLength];
                    buffer.get(bytes);
                    tempList.add(0, bytes);
                }
            } else if (redoCommitEndLengthStr.startsWith(BPlusPublicConstants.REDO_END_STR)) {
                straightForward = false;
                needRedo = false;
                String redoCommitStr = redoCommitEndLengthStr.length() > BPlusPublicConstants.REDO_END_STR.getBytes().length ?
                        redoCommitEndLengthStr.substring(BPlusPublicConstants.REDO_END_STR.getBytes().length).trim() : "";
                if (redoCommitStr.startsWith(BPlusPublicConstants.REDO_END_COMMIT_STR)) {
                    straightForward = true;
                } else {
                    needRedo = true;
                }
            } else {
                needRedo = false;
                straightForward = false;
                if (!tempList.isEmpty()) {
                    toDispose.add(tempList);
                    tempList = new ArrayList<>();
                }
            }

            position -= BPlusPublicConstants.REDO_PAGE_SIZE;
        }

        // 拼接好二进制事务列表（formedList）
        List<RedoInfo> formedList = new ArrayList<>();
        for (List<byte[]> list : toDispose) {
            int l = list.stream().mapToInt(e -> e.length).sum();
            byte[] bytes = new byte[l];
            int k = 0;
            for (byte[] value : list) {
                System.arraycopy(value, 0, bytes, k, value.length);
                k += value.length;
            }
            formedList.add(JSONObject.parseObject(bytes, RedoInfo.class));
        }

        // 倒过来，目前第一条是最后的操作，刚好反了
        List<RedoInfo> result = new ArrayList<>();
        for (int i = formedList.size() - 1; i >= 0; i--) {
            result.add(formedList.get(i));
        }

        return result;
    }

    Long[] doRedo(List<RedoInfo> formedList) {
        WRITE_LOCK.lock();
        try {
            // 执行重做
            Long[] commitTransactionIds = new Long[formedList.size()];
            for (int i = 0; i < formedList.size(); i++) {
                RedoInfo e = formedList.get(i);
                if (BPlusFileInfo.getEndPosition() < e.getEndPosition()) {
                    BPlusFileInfo.saveEndPosition(e.getEndPosition());
                }
                batchSerializeBplusTree(e.getToSerializedBplusTreeList());

                if (e.getCount() != null) {
                    BPlusFileInfo.saveCount(e.getCount());
                }
                // todo crash时这里会导致磁盘中空闲碎片较多
                BPlusFileInfo.saveEndPosition();

                if (e.getRootPosition() != null) {
                    BPlusFileInfo.saveRootPosition(e.getRootPosition());
                }
                commitTransactionIds[i] = e.getTransactionId();
            }
            return commitTransactionIds;
        } finally {
            WRITE_LOCK.unlock();
        }
    }

    private void batchSerializeBplusTree(List<BPlusTree> bPlusTreeList) {
        // 持久化b+tree
        int n = 1000;
        if (bPlusTreeList.size() >= n) {
            System.out.println("需要持久化的节点有" + bPlusTreeList.size() + "个");
        }
        int h = 0;
        long a2;
        long a1 = System.currentTimeMillis();
        for (BPlusTree m : bPlusTreeList) {
            m.serializeToFile();
            if (++h % n == 0) {
                a2 = System.currentTimeMillis();
                System.out.println("已持久化：" + h + "条数据,总共耗时：" + (a2 - a1) + "毫秒");
                a1 = a2;
            }
        }
        System.out.println("节点结束！");
    }

    static DoRedo getInstance() {
        return Singleton.INSTANCE;
    }

    private static class Singleton {
        private static final DoRedo INSTANCE = new DoRedo();
    }

    private DoRedo() {
    }
}
